package storage

import (
	"context"
	"encoding/json"
	"fmt"
	"jdfs2/internal/utils"
	"sync"
	"time"

	"github.com/dgraph-io/badger/v3"
)

// DistributedMetaStore 分布式元数据存储
type DistributedMetaStore struct {
	db           *badger.DB
	nodeID       string
	validator    *MetaValidator
	compressor   *MetaCompressor
	metrics      *MetaMetrics
	mu           sync.RWMutex
	lastSyncTime time.Time
}

// NewDistributedMetaStore 创建新的分布式元数据存储
func NewDistributedMetaStore(path, nodeID string) (*DistributedMetaStore, error) {
	opts := badger.DefaultOptions(path)
	opts.Logger = nil // 禁用日志以减少开销

	db, err := badger.Open(opts)
	if err != nil {
		return nil, fmt.Errorf("failed to open badger db: %w", err)
	}

	return &DistributedMetaStore{
		db:         db,
		nodeID:     nodeID,
		validator:  NewMetaValidator(),
		compressor: NewMetaCompressor(6),
		metrics:    &MetaMetrics{},
	}, nil
}

// PutMeta 存储元数据
func (dms *DistributedMetaStore) PutMeta(bucket, key string, meta *DistributedObjectMeta) error {
	// 验证元数据
	if err := dms.validator.Validate(meta); err != nil {
		return fmt.Errorf("meta validation failed: %w", err)
	}

	// 更新元数据时间戳
	meta.UpdatedAt = time.Now().Unix()
	if meta.CreatedAt == 0 {
		meta.CreatedAt = meta.UpdatedAt
	}

	// 序列化元数据
	data, err := meta.Serialize()
	if err != nil {
		return fmt.Errorf("failed to serialize meta: %w", err)
	}

	// 存储到数据库
	k := dms.metaKey(bucket, key)
	return dms.db.Update(func(tx *badger.Txn) error {
		return tx.Set(k, data)
	})
}

// GetMeta 获取元数据
func (dms *DistributedMetaStore) GetMeta(bucket, key string) (*DistributedObjectMeta, error) {
	k := dms.metaKey(bucket, key)
	var meta DistributedObjectMeta

	err := dms.db.View(func(tx *badger.Txn) error {
		item, err := tx.Get(k)
		if err != nil {
			if err == badger.ErrKeyNotFound {
				return utils.ErrNoSuchKey
			}
			return err
		}

		return item.Value(func(val []byte) error {
			return meta.Deserialize(val)
		})
	})

	if err != nil {
		return nil, err
	}

	// 检查是否已删除
	if meta.IsDeleted() {
		return nil, utils.ErrNoSuchKey
	}

	return &meta, nil
}

// DeleteMeta 删除元数据
func (dms *DistributedMetaStore) DeleteMeta(bucket, key string) error {
	// 获取现有元数据
	meta, err := dms.GetMeta(bucket, key)
	if err != nil {
		if err == utils.ErrNoSuchKey {
			return nil // 删除不存在的对象是幂等的
		}
		return err
	}

	// 标记为已删除而不是真正删除
	meta.MarkDeleted()
	return dms.PutMeta(bucket, key, meta)
}

// ListMeta 列出元数据
func (dms *DistributedMetaStore) ListMeta(bucket, prefix string, limit int) ([]*DistributedObjectMeta, error) {
	var metas []*DistributedObjectMeta
	searchPrefix := dms.metaKey(bucket, prefix)

	err := dms.db.View(func(tx *badger.Txn) error {
		it := tx.NewIterator(badger.DefaultIteratorOptions)
		defer it.Close()

		count := 0
		for it.Seek(searchPrefix); it.ValidForPrefix(searchPrefix) && count < limit; it.Next() {
			item := it.Item()
			var meta DistributedObjectMeta

			err := item.Value(func(val []byte) error {
				return meta.Deserialize(val)
			})
			if err != nil {
				return err
			}

			// 跳过已删除的对象
			if !meta.IsDeleted() {
				metas = append(metas, &meta)
				count++
			}
		}
		return nil
	})

	return metas, err
}

// BatchPutMeta 批量存储元数据
func (dms *DistributedMetaStore) BatchPutMeta(operations []MetaOperation) error {
	return dms.db.Update(func(tx *badger.Txn) error {
		for _, op := range operations {
			if op.Meta == nil {
				continue
			}

			// 验证元数据
			if err := dms.validator.Validate(op.Meta); err != nil {
				return fmt.Errorf("meta validation failed for %s/%s: %w", op.Bucket, op.Key, err)
			}

			// 序列化元数据
			data, err := op.Meta.Serialize()
			if err != nil {
				return fmt.Errorf("failed to serialize meta for %s/%s: %w", op.Bucket, op.Key, err)
			}

			// 存储到数据库
			k := dms.metaKey(op.Bucket, op.Key)
			if err := tx.Set(k, data); err != nil {
				return err
			}
		}
		return nil
	})
}

// GetMetaBatch 获取批量元数据
func (dms *DistributedMetaStore) GetMetaBatch(bucket string, keys []string) (map[string]*DistributedObjectMeta, error) {
	metas := make(map[string]*DistributedObjectMeta)

	err := dms.db.View(func(tx *badger.Txn) error {
		for _, key := range keys {
			k := dms.metaKey(bucket, key)
			item, err := tx.Get(k)
			if err != nil {
				if err == badger.ErrKeyNotFound {
					continue // 跳过不存在的键
				}
				return err
			}

			var meta DistributedObjectMeta
			err = item.Value(func(val []byte) error {
				return meta.Deserialize(val)
			})
			if err != nil {
				return err
			}

			// 跳过已删除的对象
			if !meta.IsDeleted() {
				metas[key] = &meta
			}
		}
		return nil
	})

	return metas, err
}

// SyncMeta 同步元数据 (用于分布式同步)
func (dms *DistributedMetaStore) SyncMeta(batch *MetaBatch) error {
	dms.mu.Lock()
	defer dms.mu.Unlock()

	// 验证批量操作
	if batch == nil || len(batch.Operations) == 0 {
		return nil
	}

	// 应用批量操作
	return dms.BatchPutMeta(batch.Operations)
}

// GetMetaSince 获取指定时间后的元数据变更
func (dms *DistributedMetaStore) GetMetaSince(bucket string, since int64) ([]*DistributedObjectMeta, error) {
	var metas []*DistributedObjectMeta
	searchPrefix := dms.metaKey(bucket, "")

	err := dms.db.View(func(tx *badger.Txn) error {
		it := tx.NewIterator(badger.DefaultIteratorOptions)
		defer it.Close()

		for it.Seek(searchPrefix); it.ValidForPrefix(searchPrefix); it.Next() {
			item := it.Item()
			var meta DistributedObjectMeta

			err := item.Value(func(val []byte) error {
				return meta.Deserialize(val)
			})
			if err != nil {
				return err
			}

			// 只返回指定时间后的变更
			if meta.UpdatedAt > since {
				metas = append(metas, &meta)
			}
		}
		return nil
	})

	return metas, err
}

// GetMetaStats 获取元数据统计信息
func (dms *DistributedMetaStore) GetMetaStats() (*MetaStats, error) {
	stats := &MetaStats{
		NodeID: dms.nodeID,
	}

	err := dms.db.View(func(tx *badger.Txn) error {
		it := tx.NewIterator(badger.DefaultIteratorOptions)
		defer it.Close()

		for it.Rewind(); it.Valid(); it.Next() {
			item := it.Item()
			key := string(item.Key())

			// 跳过非元数据键
			if !dms.isMetaKey(key) {
				continue
			}

			stats.TotalKeys++

			var meta DistributedObjectMeta
			err := item.Value(func(val []byte) error {
				return meta.Deserialize(val)
			})
			if err != nil {
				continue // 跳过损坏的数据
			}

			if meta.IsDeleted() {
				stats.DeletedKeys++
			} else {
				stats.ActiveKeys++
				stats.TotalSize += meta.Size
				stats.TotalReplicas += len(meta.Replicas)
			}
		}
		return nil
	})

	stats.LastSyncTime = dms.lastSyncTime
	return stats, err
}

// CompactMeta 压缩元数据 (清理已删除的对象)
func (dms *DistributedMetaStore) CompactMeta(bucket string, beforeTime int64) error {
	searchPrefix := dms.metaKey(bucket, "")

	return dms.db.Update(func(tx *badger.Txn) error {
		it := tx.NewIterator(badger.DefaultIteratorOptions)
		defer it.Close()

		for it.Seek(searchPrefix); it.ValidForPrefix(searchPrefix); it.Next() {
			item := it.Item()
			var meta DistributedObjectMeta

			err := item.Value(func(val []byte) error {
				return meta.Deserialize(val)
			})
			if err != nil {
				continue // 跳过损坏的数据
			}

			// 删除指定时间前的已删除对象
			if meta.IsDeleted() && meta.DeletedAt < beforeTime {
				if err := tx.Delete(item.KeyCopy(nil)); err != nil {
					return err
				}
			}
		}
		return nil
	})
}

// GetBucketConfig 获取桶配置
func (dms *DistributedMetaStore) GetBucketConfig(bucket string) (*BucketConfig, error) {
	var cfg BucketConfig
	k := dms.bucketConfigKey(bucket)

	err := dms.db.View(func(tx *badger.Txn) error {
		item, err := tx.Get(k)
		if err != nil {
			if err == badger.ErrKeyNotFound {
				return nil // 返回默认配置
			}
			return err
		}
		return item.Value(func(val []byte) error {
			return json.Unmarshal(val, &cfg)
		})
	})

	return &cfg, err
}

// PutBucketConfig 存储桶配置
func (dms *DistributedMetaStore) PutBucketConfig(bucket string, cfg *BucketConfig) error {
	k := dms.bucketConfigKey(bucket)
	v, err := json.Marshal(cfg)
	if err != nil {
		return err
	}
	return dms.db.Update(func(tx *badger.Txn) error {
		return tx.Set(k, v)
	})
}

// DeleteBucketConfig 删除桶配置
func (dms *DistributedMetaStore) DeleteBucketConfig(bucket string) error {
	k := dms.bucketConfigKey(bucket)
	return dms.db.Update(func(tx *badger.Txn) error {
		err := tx.Delete(k)
		if err == badger.ErrKeyNotFound {
			return nil
		}
		return err
	})
}

// PutDeletedRecord 保存已删除对象的记录
func (dms *DistributedMetaStore) PutDeletedRecord(bucket, key string, record *DeletedObjectRecord) error {
	k := dms.deletedRecordKey(bucket, key)
	v, err := json.Marshal(record)
	if err != nil {
		return err
	}
	return dms.db.Update(func(tx *badger.Txn) error {
		return tx.Set(k, v)
	})
}

// DeleteAllMetaInBucket 删除桶中所有元数据
func (dms *DistributedMetaStore) DeleteAllMetaInBucket(bucket string) error {
	searchPrefix := dms.metaKey(bucket, "")
	return dms.db.Update(func(tx *badger.Txn) error {
		it := tx.NewIterator(badger.DefaultIteratorOptions)
		defer it.Close()
		for it.Seek(searchPrefix); it.ValidForPrefix(searchPrefix); it.Next() {
			item := it.Item()
			if err := tx.Delete(item.KeyCopy(nil)); err != nil {
				return err
			}
		}
		return nil
	})
}

// ListDeletedRecordsByBucket 列出指定桶的已删除对象记录
func (dms *DistributedMetaStore) ListDeletedRecordsByBucket(bucket string) ([]DeletedObjectRecord, error) {
	var records []DeletedObjectRecord
	prefix := dms.deletedRecordKey(bucket, "")

	err := dms.db.View(func(tx *badger.Txn) error {
		it := tx.NewIterator(badger.DefaultIteratorOptions)
		defer it.Close()

		for it.Seek(prefix); it.ValidForPrefix(prefix); it.Next() {
			item := it.Item()
			var record DeletedObjectRecord

			err := item.Value(func(val []byte) error {
				return json.Unmarshal(val, &record)
			})
			if err != nil {
				return err
			}

			records = append(records, record)
		}
		return nil
	})

	return records, err
}

// Close 关闭存储
func (dms *DistributedMetaStore) Close() error {
	if dms.db != nil {
		return dms.db.Close()
	}
	return nil
}

// metaKey 生成元数据键
func (dms *DistributedMetaStore) metaKey(bucket, key string) []byte {
	return []byte(fmt.Sprintf("meta:%s:%s", bucket, key))
}

// bucketConfigKey 生成桶配置键
func (dms *DistributedMetaStore) bucketConfigKey(bucket string) []byte {
	return []byte(fmt.Sprintf("bucket_config:%s", bucket))
}

// deletedRecordKey 生成删除记录的键
func (dms *DistributedMetaStore) deletedRecordKey(bucket, key string) []byte {
	return []byte(fmt.Sprintf("deleted:%s:%s", bucket, key))
}

// isMetaKey 检查是否为元数据键
func (dms *DistributedMetaStore) isMetaKey(key string) bool {
	return len(key) > 5 && key[:5] == "meta:"
}

// MetaStats 元数据统计信息
type MetaStats struct {
	NodeID        string    `json:"node_id"`
	TotalKeys     int64     `json:"total_keys"`
	ActiveKeys    int64     `json:"active_keys"`
	DeletedKeys   int64     `json:"deleted_keys"`
	TotalSize     int64     `json:"total_size"`
	TotalReplicas int       `json:"total_replicas"`
	LastSyncTime  time.Time `json:"last_sync_time"`
}

// DistributedMetaStoreConfig 分布式元数据存储配置
type DistributedMetaStoreConfig struct {
	NodeID           string        `yaml:"node_id"`
	CompressionLevel int           `yaml:"compression_level"`
	SyncInterval     time.Duration `yaml:"sync_interval"`
	CompactInterval  time.Duration `yaml:"compact_interval"`
	MaxBatchSize     int           `yaml:"max_batch_size"`
}

// NewDistributedMetaStoreWithConfig 使用配置创建分布式元数据存储
func NewDistributedMetaStoreWithConfig(path string, config *DistributedMetaStoreConfig) (*DistributedMetaStore, error) {
	store, err := NewDistributedMetaStore(path, config.NodeID)
	if err != nil {
		return nil, err
	}

	// 应用配置
	if config.CompressionLevel > 0 {
		store.compressor = NewMetaCompressor(config.CompressionLevel)
	}

	return store, nil
}

// MetaStoreInterface 元数据存储接口 (用于向后兼容)
type MetaStoreInterface interface {
	PutMeta(bucket, key string, meta *DistributedObjectMeta) error
	GetMeta(bucket, key string) (*DistributedObjectMeta, error)
	DeleteMeta(bucket, key string) error
	ListMeta(bucket, prefix string, limit int) ([]*DistributedObjectMeta, error)
	BatchPutMeta(operations []MetaOperation) error
	GetMetaBatch(bucket string, keys []string) (map[string]*DistributedObjectMeta, error)
	SyncMeta(batch *MetaBatch) error
	GetMetaSince(bucket string, since int64) ([]*DistributedObjectMeta, error)
	GetMetaStats() (*MetaStats, error)
	CompactMeta(bucket string, beforeTime int64) error
	Close() error
}

// MetaStoreContext 元数据存储上下文
type MetaStoreContext struct {
	Store   MetaStoreInterface
	Context context.Context
	Cancel  context.CancelFunc
	Metrics *MetaMetrics
	Config  *DistributedMetaStoreConfig
}

// NewMetaStoreContext 创建元数据存储上下文
func NewMetaStoreContext(store MetaStoreInterface, config *DistributedMetaStoreConfig) *MetaStoreContext {
	ctx, cancel := context.WithCancel(context.Background())

	return &MetaStoreContext{
		Store:   store,
		Context: ctx,
		Cancel:  cancel,
		Metrics: &MetaMetrics{},
		Config:  config,
	}
}

// StartBackgroundTasks 启动后台任务
func (ctx *MetaStoreContext) StartBackgroundTasks() {
	// 启动同步任务
	go ctx.syncTask()

	// 启动压缩任务
	go ctx.compactTask()

	// 启动指标收集任务
	go ctx.metricsTask()
}

// Stop 停止后台任务
func (ctx *MetaStoreContext) Stop() {
	ctx.Cancel()
}

// syncTask 同步任务
func (ctx *MetaStoreContext) syncTask() {
	ticker := time.NewTicker(ctx.Config.SyncInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Context.Done():
			return
		case <-ticker.C:
			// 执行同步逻辑
			ctx.performSync()
		}
	}
}

// compactTask 压缩任务
func (ctx *MetaStoreContext) compactTask() {
	ticker := time.NewTicker(ctx.Config.CompactInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Context.Done():
			return
		case <-ticker.C:
			// 执行压缩逻辑
			ctx.performCompact()
		}
	}
}

// metricsTask 指标收集任务
func (ctx *MetaStoreContext) metricsTask() {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Context.Done():
			return
		case <-ticker.C:
			// 收集指标
			ctx.collectMetrics()
		}
	}
}

// performSync 执行同步
func (ctx *MetaStoreContext) performSync() {
	// 这里实现具体的同步逻辑
	// 例如：从其他节点同步元数据
}

// performCompact 执行压缩
func (ctx *MetaStoreContext) performCompact() {
	// 这里实现具体的压缩逻辑
	// 例如：清理已删除的元数据
}

// collectMetrics 收集指标
func (ctx *MetaStoreContext) collectMetrics() {
	// 这里实现具体的指标收集逻辑
	// 例如：收集性能指标和统计信息
}
